package com.bw.gmall.realtime.dws.app;

import com.bw.gmall.realtime.common.base.BaseSqlApp;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.SQLUtil;
import com.bw.gmall.realtime.dws.function.KwSplit;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @projectName: gmalll2206
 * @package: com.bw.gmall.realtime.dws.app
 * @className: DwsTrafficSourceKeywordPageViewWindow
 * @author: DuanYu
 * @description: TODO
 * @date: 2025/2/28 11:24
 * @version: 1.0
 */

public class DwsTrafficSourceKeywordPageViewWindow extends BaseSqlApp {
    public static void main(String[] args) {
        new DwsTrafficSourceKeywordPageViewWindow().start(Constant.DWS_TRAFFIC_SOURCE_KEYWORD_PAGE_VIEW_WINDOW, Constant.DWS_TRAFFIC_VC_CH_AR_IS_NEW_PAGE_VIEW_WINDOW, 4, 10021);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv, String groupId) {
        //读取page_log的信息
        TableResult dws_traffic_source_keyword_page_view_window = getDwsTrafficSourceKeywordPageViewWindow(tableEnv);
//        tableEnv.sqlQuery("select * from page_log ").execute().print();

        //获取item
        Table item = getItem(tableEnv);
//        item.execute().print();

        tableEnv.createTemporaryView("kw_table", item);

        // 3. 自定义分词函数
        tableEnv.createTemporaryFunction("kw_split", KwSplit.class);
        Table keyword = getKeyword(tableEnv);
//        keyword.execute().print();
        tableEnv.createTemporaryView("keyword_table", keyword);
        // 4开窗聚和 tvf
        Table table = keyWordCount(tableEnv);
//        table.execute().print();
        tableEnv.executeSql("create table "+ Constant.DWS_TRAFFIC_SOURCE_KEYWORD_PAGE_VIEW_WINDOW+"\n" +
                "(\n" +
                "    `start_time`           STRING COMMENT '窗口起始时间',\n" +
                "    `end_time`           STRING COMMENT '窗口结束时间',\n" +
                "    `cur_date`      STRING COMMENT '当天日期',\n" +
                "    `keyword`       STRING COMMENT '搜索关键词',\n" +
                "    `keyword_count` BIGINT  COMMENT '搜索关键词出现次数'\n" +
                "    )"+SQLUtil.getDorisSinkSQL(Constant.DWS_TRAFFIC_SOURCE_KEYWORD_PAGE_VIEW_WINDOW));
        table.executeInsert(Constant.DWS_TRAFFIC_SOURCE_KEYWORD_PAGE_VIEW_WINDOW);

    }

    private static Table keyWordCount(StreamTableEnvironment tableEnv) {
        return tableEnv.sqlQuery("select date_format(window_start,'yyyy-MM-dd HH:mm:ss') window_start,\n" +
                "       date_format(window_end,'yyyy-MM-dd HH:mm:ss') window_end,\n" +
                "       date_format(now(), 'yyyyMMdd') cur_date,\n" +
                "       keyword,\n" +
                "       count(*) keyword_count\n" +
                " FROM TABLE(\n" +
                "    TUMBLE(TABLE keyword_table, DESCRIPTOR(et), INTERVAL '5' SECOND))\n" +
                "GROUP BY window_start, window_end,keyword;");
    }

    private static Table getKeyword(StreamTableEnvironment tableEnv) {
        return tableEnv.sqlQuery("select keyword, et\n" +
                "from kw_table\n" +
                "         join lateral table(kw_split(kw))\n" +
                "on true;");
    }

    private static Table getItem(StreamTableEnvironment tableEnv) {
        return tableEnv.sqlQuery("select \n" +
                "    page['item'] kw ,\n" +
                "    et \n" +
                "from page_log\n" +
                "where (page['last_page_id'] = 'search'\n" +
                "or page['last_page_id'] = 'home')\n" +
                "and page['item_type'] = 'keyword'\n" +
                "and page['item'] is not null");
    }

    private static TableResult getDwsTrafficSourceKeywordPageViewWindow(StreamTableEnvironment tableEnv) {
        return tableEnv.executeSql("create table page_log(" +
                " page map<string, string>, " +
                " ts bigint, " +
                " et as to_timestamp_ltz(ts, 3), " +
                " watermark for et as et - interval '5' second " +
                ")" + SQLUtil.getKafkaSourceSQL(Constant.TOPIC_DWD_TRAFFIC_PAGE, Constant.TOPIC_DWD_TRAFFIC_PAGE));

    }
}
